package com.gjy.learning.scala

import com.gjy.learning.scala.SparkReadOracle.oracleDFCreate
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.col

object BroadCastDemo {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("Broadcast Join Example")
      .master("local[*]")
      .getOrCreate()

    // 创建小数据集
    var smallData = BigDecimal(10)

    // 广播小数据集
    var smallDataBroadcast: Broadcast[BigDecimal] = spark.sparkContext.broadcast(smallData)
    // 创建大数据集
    val emp = oracleDFCreate(spark, "emp")

    emp.show

    emp.filter(col("deptno") === smallDataBroadcast.value).show()
    // 修改小数据集
    smallData = BigDecimal(20)

    emp.filter(col("deptno") === smallDataBroadcast.value).show()// 没有变化

    // 重新广播小数据集
    smallDataBroadcast = spark.sparkContext.broadcast(smallData)

    emp.filter(col("deptno") === smallDataBroadcast.value).show()// 变化

    //释放广播变量内存
    smallDataBroadcast.unpersist()
    // 关闭 Spark 会话
    spark.stop()
  }

}

